#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
#include "server.h"
#include "log.h"

static SERVERINFO *server;
static WORK_THREAD *work_threads;
static int init_count = 0;
static pthread_mutex_t init_lock;
static pthread_cond_t init_cond;
static int round_robin = 0;
static int logmode ;
static int loglevel;
static char logfile[256] = {0};
static int exitednum = 0;


static int socket_init();
static void accept_callback(struct ev_loop *loop, ev_io *w, int revents);
static void recv_callback(struct ev_loop *loop, ev_io *w, int revents);
static void write_callback(struct ev_loop *loop, ev_io *w, int revents);
static void signal_callback(struct ev_loop *loop, ev_signal *w, int revents);


SERVERINFO * create_server()
{
   server = (SERVERINFO *)calloc(1,sizeof(SERVERINFO)) ; 
   if (NULL == server){
      WRITE_ERROR("server created failed!");
   }
    
   return server ;
}

int set_serverinfo(SERVERINFO *s,int typemask,void *data)
{
  if (NULL == data) return -1;

  switch (typemask){
    case T_PORT:
      s->port = *(int *)data ;
      break;
    case T_WORKTHREAD_NUM:
      s->workthread_num = *(int *)data ;
      break;
    case T_READBUFFERSIZE:
      s->rbuffersize = *(int *)data ;
      break;
    case T_WRITEBUFFERSIZE:
      s->wbuffersize = *(int *)data ;
      break;
    case T_MAXCONNS:
      s->maxconns = *(int *)data;
      break;
    case T_INIT_USER_DATA_FN:
      s->pfn_init_user_data = (init_user_data *)data;
      break;
    case T_FREE_USER_DATA_FN:
      s->pfn_free_user_data = (free_user_data *)data;
      break;
    case T_DATA_CALLBACK:
      s->pfn_network_data_callback = (network_data_callback *)data;
      break;
    default:
      break;
  }
  
  return 0;
}

static CONN_ITEM *get_conn_byfd(int fd,WORK_THREAD *wt)
{
   if (NULL == wt) return NULL;
   if (NULL == wt->connlist) return NULL ;

   char strfd[8]={0};
   sprintf(strfd,"%d",fd);
   CONN_ITEM *c = (CONN_ITEM *)hash_value(strfd,wt->connlist);
   return c;
}

void set_loginfo(int mode,int level,char *log)
{
  initlog(mode ,level,log) ;
}


void setnonblocking(int fd)
{
  int opts;
  opts = fcntl (fd,F_GETFL);
  if (opts < 0)
  {
    perror("fcntl(fd,GETTFL)");
    exit(1);
  }

  opts = opts | O_NONBLOCK;
  if (fcntl(fd,F_SETFL,opts) <0 )
  {
    perror("fcntl(fd,SETFL,opts)");
    exit(1);
  }
}

static WORK_THREAD *thisworkthread()
{
   int i = 0;
   for (;i<server->workthread_num && (unsigned int)work_threads[i].thread_id != (unsigned int)pthread_self();i++);
   if (i == server->workthread_num) return NULL;
   return &work_threads[i];
}

static int total_conns()
{
  int i = 0,total = 0;
  for (;i<server->workthread_num;i++)
     total+=hash_count(work_threads[i].connlist);
  return total;
}

static void create_connection(WORK_THREAD *wt,int sfd,char *clientip,int nport)
{
   WRITE_INFO("create connection :[%lu]: %d [%s:%d]",pthread_self(),
       sfd,clientip,nport);
   
   CONN_ITEM *c = (CONN_ITEM *)calloc(1,sizeof(CONN_ITEM));
   c->sfd = sfd ;
   c->readbuffer = (char *)calloc(1,server->rbuffersize); 
   c->writebuffer = (char *)calloc(1,server->wbuffersize);
   c->rlastpos = 0;
   c->wlastpos = 0;
   c->rbuffersize = server->rbuffersize ;
   c->wbuffersize = server->wbuffersize ;
   strcpy(c->clientip,clientip);
   c->port = nport ;
   pthread_mutex_init(&c->conn_locker,NULL);
   if (NULL != server->pfn_init_user_data)
     server->pfn_init_user_data(c);
   setnonblocking(c->sfd);

   ev_io_init(&c->io_watcher,recv_callback,c->sfd,EV_READ);
   ev_io_start(wt->loop,&c->io_watcher);

   char strfd[8]={0};
   sprintf(strfd,"%d",sfd); 
   assert(NULL != wt);
   if (NULL == wt || NULL == wt->connlist) WRITE_DEBUG("wt IS NULL");
   hash_insert((void *)strdup(strfd),(void *)c,wt->connlist);
   int totalconns = total_conns();
   WRITE_DEBUG("CONNS:%d",totalconns);
}

static void free_connection(CONN_ITEM *c)
{

   char strfd[8]={0};
   sprintf(strfd,"%d",c->sfd);

#if 0
   CONN_ITEM *c = (CONN_ITEM *)hash_value(strfd,thisworkthread()->connlist);
   if (NULL == c){
     pthread_mutex_unlock(&thisworkthread()->connlist_locker);
     WRITE_ERROR("free_connection() CONN_ITEM is NULL"); 
     return ;
   }
#endif

   WRITE_INFO("closing %d [%s:%d]",c->sfd,c->clientip,c->port);

   pthread_mutex_lock(&c->conn_locker);
   if (NULL != c->readbuffer){
     free(c->readbuffer);
     c->readbuffer = NULL ;
   }
   
   if (NULL != c->writebuffer){
     free(c->writebuffer);
     c->writebuffer = NULL ;
   }

   if (NULL != c->userdata && NULL != server->pfn_free_user_data){
     server->pfn_free_user_data(c->userdata);
   }
   pthread_mutex_unlock(&c->conn_locker);
   pthread_mutex_destroy(&c->conn_locker);

   if (NULL != server->pfn_free_user_data) 
      server->pfn_free_user_data(c);
   
   pthread_mutex_lock(&thisworkthread()->connlist_locker);
   hash_remove((void *)strfd,thisworkthread()->connlist);
   pthread_mutex_unlock(&thisworkthread()->connlist_locker);

   round_robin--;
  
   WORK_THREAD *wt = thisworkthread();
   if (NULL != wt && wt->ifexit == 0) 
      ev_io_stop(wt->loop,&c->io_watcher);
   
   close(c->sfd); 
   c->sfd = -1;
   WRITE_DEBUG("CONNS:%d",total_conns());
}



static int connlist_each_do(void *key,void *data)
{
   WRITE_DEBUG("enter connlist_each_do");
   free_connection((CONN_ITEM *)data);
 
   return 0 ;
}

/*
 * Worker thread: main event loop
 */
static void *worker_libev(void *arg) {
    WORK_THREAD *me = arg;


    /* Any per-thread setup can happen here; thread_init() will block until
     * all threads have finished initializing.
     */


    pthread_mutex_lock(&init_lock);
    init_count++;
    pthread_cond_signal(&init_cond);
    pthread_mutex_unlock(&init_lock);


    me->thread_id = pthread_self();
    WRITE_DEBUG("me->thread_id:%u",(unsigned int)me->thread_id);
    ev_loop(me->loop, 0);
    ev_loop_destroy(me->loop);
      
    pthread_cond_signal(&me->pending_queue_cond);

    WRITE_DEBUG("prepare hash_each_do");
    hash_for_each_do(me->connlist,connlist_each_do);
    
    exitednum++;

    return NULL;
}


static void *logic_worker(void *arg){
  WORK_THREAD *wt = arg;
  
  WRITE_INFO("logic_worker %lu started",pthread_self());  
  
  while ( !wt->ifexit ){

    pthread_mutex_lock(&wt->pending_queue_locker);
    struct pending_queue_item *item = NULL;
    while(!(item = TAILQ_FIRST(&wt->pending_queue)) && !wt->ifexit){
      pthread_cond_wait(&wt->pending_queue_cond,&wt->pending_queue_locker);
    }

    while(item = TAILQ_FIRST(&wt->pending_queue)){
       CONN_ITEM *c = get_conn_byfd(item->sfd,wt);
       TAILQ_REMOVE(&wt->pending_queue,item,entries);
       free(item);
       item = NULL ;

       if (NULL == c){
         continue;
       }

       pthread_mutex_lock(&c->conn_locker);
       while (c->rlastpos>=sizeof(int)){
          int datalen = *(int *)c->readbuffer;
          int pkgsize = datalen + sizeof(int);
          if (server->pfn_network_data_callback && c->rlastpos>=pkgsize)
             server->pfn_network_data_callback(wt->loop,c,c->readbuffer+sizeof(int),datalen);
          if  (c->rbuffersize>pkgsize)
             memmove(c->readbuffer,c->readbuffer+pkgsize,c->rbuffersize - pkgsize);
          
          c->rlastpos -= pkgsize;
       }
      pthread_mutex_unlock(&c->conn_locker);
      
      if (c->wlastpos == 0){
        struct pending_queue_item *lpNewItem = calloc(1,sizeof(struct pending_queue_item));
        if (! lpNewItem) {
          perror("Can't allocate connection item\n");
          exit(1);
        }
        lpNewItem->sfd = c->sfd;
        lpNewItem->port = -1;
        pthread_mutex_lock(&wt->new_conn_queue_locker);
        TAILQ_INSERT_TAIL(&wt->new_conn_queue,lpNewItem,entries);
        pthread_mutex_unlock(&wt->new_conn_queue_locker);
        ev_async_send(wt->loop, &wt->async_watcher);
      }
    }
    pthread_mutex_unlock(&wt->pending_queue_locker);
  } 

  pthread_mutex_destroy(&wt->connlist_locker);
  pthread_mutex_destroy(&wt->pending_queue_locker);
  pthread_cond_destroy(&wt->pending_queue_cond);
  pthread_mutex_destroy(&wt->new_conn_queue_locker);

  exitednum++;
  WRITE_INFO("logic_worker %lu terminated",pthread_self());  
}


/*
 * Creates a worker thread.
 */
static void create_worker(void *(*func)(void *), void *arg) {
    pthread_t       thread;
    pthread_attr_t  attr;
    int             ret;


    pthread_attr_init(&attr);


    if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) {
        fprintf(stderr, "Can't create thread: %s\n",
                strerror(ret));
        exit(1);
    }
}


static void
async_cb (EV_P_ ev_async *w, int revents)
{
    WORK_THREAD *wt = (WORK_THREAD *)(w->data) ;
    if (wt->ifexit){
      WRITE_DEBUG("thread[%lu] prepare terminated",pthread_self());
      ev_break(loop,EVBREAK_ALL);
      return ;
    }

    pthread_mutex_lock(&wt->new_conn_queue_locker);
    struct pending_queue_item *item = TAILQ_FIRST(&wt->new_conn_queue); 

    if (NULL != item) { 
      if (item->port > 0){       
        create_connection(wt,item->sfd,item->clientip,item->port);
 
        WRITE_INFO("thread[%lu] accept: fd :%d  addr:%s port:%d\n",wt->thread_id,item->sfd,item->clientip,item->port);
      }else{
        CONN_ITEM *c = get_conn_byfd(item->sfd,wt);
        ev_io_stop(wt->loop,&c->io_watcher);
        ev_io_init(&c->io_watcher,recv_callback,c->sfd,(item->port==-1)?EV_READ:EV_WRITE);
        ev_io_start(wt->loop,&c->io_watcher);
      }
      TAILQ_REMOVE(&wt->new_conn_queue,item,entries);
      free(item);
      item = NULL;
    }
    pthread_mutex_unlock(&wt->new_conn_queue_locker);
}


/*
 * Set up a thread's information.
 */
static void setup_thread(WORK_THREAD *me) {
   WRITE_DEBUG("server->maxconns:%d",server->maxconns);
   me->connlist = create_hashtable(server->maxconns);
   if (NULL == me->connlist){
     WRITE_ERROR("connlist create failed");
     return ;
   }

    me->loop = ev_loop_new(0);
    if (! me->loop) {
        fprintf(stderr, "Can't allocate event base\n");
        exit(1);
    }

    me->async_watcher.data = me;
    /* Listen for notifications from other threads */
    ev_async_init(&me->async_watcher, async_cb);
    ev_async_start(me->loop, &me->async_watcher);
    me->ifexit = 0;
    ev_set_userdata(me->loop,me);
    TAILQ_INIT(&me->new_conn_queue);
    TAILQ_INIT(&me->pending_queue);
    pthread_mutex_init(&me->new_conn_queue_locker,NULL);
    pthread_mutex_init(&me->pending_queue_locker,NULL);
    pthread_cond_init(&me->pending_queue_cond,NULL); 
    pthread_mutex_init(&me->connlist_locker,NULL);
}


void thread_init()
{
    int nthreads = server->workthread_num ;
    pthread_mutex_init(&init_lock, NULL);
    pthread_cond_init(&init_cond, NULL);
    work_threads = calloc(server->workthread_num, sizeof(WORK_THREAD));
    if (! work_threads) {
        perror("Can't allocate thread descriptors\n");
        exit(1);
    }

    
   server->thread_id = pthread_self();
   server->loop = ev_default_loop(0);
    
    int i = 0;
    for (i = 0; i < server->workthread_num; i++) {
        setup_thread(&work_threads[i]);
    }

    for (i = 0; i < server->workthread_num; i++) {
        create_worker(logic_worker, &work_threads[i]);
    }

    /* Create threads after we've done all the libevent setup. */
    for (i = 0; i < server->workthread_num; i++) {
        create_worker(worker_libev, &work_threads[i]);
    }


    /* Wait for all the threads to set themselves up before returning. */
    pthread_mutex_lock(&init_lock);
    while (init_count < nthreads) {
        pthread_cond_wait(&init_cond, &init_lock);
    }
    pthread_mutex_unlock(&init_lock);
}

int start(SERVERINFO *s)
{
   thread_init();

   int listen;
   listen=socket_init();
   ev_io_init(&(s->accept_watcher), accept_callback,listen, EV_READ);
   ev_io_start(s->loop,&(s->accept_watcher)); 
   ev_signal_init(&(s->signal_watcher),signal_callback,SIGTERM);
   ev_signal_start(s->loop,&(s->signal_watcher));
   ev_loop(s->loop,0);
   ev_loop_destroy(s->loop);
   return listen;
}

int stop(SERVERINFO *s)
{ 
   return 0;
}

int socket_init()
{
   struct sockaddr_in my_addr;
   int listener;
   if ((listener = socket(AF_INET, SOCK_STREAM, 0)) == -1) 
   {
     perror("socket");
     exit(1);
   } 
   else
   {
     printf("SOCKET CREATE SUCCESS! \n");
   }
   //setnonblocking(listener);
   int so_reuseaddr=1;
   setsockopt(listener,SOL_SOCKET,SO_REUSEADDR,&so_reuseaddr,sizeof(so_reuseaddr));
   bzero(&my_addr, sizeof(my_addr));
   my_addr.sin_family = AF_INET;
   my_addr.sin_port = htons(server->port);
   my_addr.sin_addr.s_addr = INADDR_ANY;


   if (bind(listener, (struct sockaddr *) &my_addr, sizeof(struct sockaddr))== -1) 
   {
     perror("bind error!\n");
     exit(1);
   } 
   else
   {
     printf("IP BIND SUCCESS,IP\n");
   }


   if (listen(listener, 1024) == -1) 
   {
     perror("listen error!\n");
     exit(1);
   } 
   else
   {
     printf("LISTEN SUCCESS,PORT:%d\n",server->port);
   }
   return listener;
}


void dispath_conn(int anewfd,struct sockaddr_in asin)
{
  // set the new connect item
  struct pending_queue_item *lpNewItem = calloc(1,sizeof(struct pending_queue_item));
  if (! lpNewItem) {
     perror("Can't allocate connection item\n");
     exit(1);
  }
  
  lpNewItem->sfd = anewfd;
  strcpy(lpNewItem->clientip,(char*)inet_ntoa(asin.sin_addr));
  lpNewItem->port = asin.sin_port;

  // libev default loop, accept the new connection, round-robin 
  // dispath to a work_thread.
  int robin = round_robin%init_count;
  pthread_mutex_lock(&work_threads[robin].new_conn_queue_locker);
  TAILQ_INSERT_TAIL(&work_threads[robin].new_conn_queue,lpNewItem,entries);
  pthread_mutex_unlock(&work_threads[robin].new_conn_queue_locker);
  ev_async_send(work_threads[robin].loop, &(work_threads[robin].async_watcher));
  round_robin++;
} 


void accept_callback(struct ev_loop *loop, ev_io *w, int revents)
{
   int newfd;
   struct sockaddr_in sin;
   socklen_t addrlen = sizeof(struct sockaddr);
   while ((newfd = accept(w->fd, (struct sockaddr *)&sin, &addrlen)) < 0)
   {
     if (errno == EAGAIN || errno == EWOULDBLOCK) 
     {
       //these are transient, so don't log anything.
       continue; 
     }
     else
     {
       WRITE_ERROR("accept error.[%s]\n", strerror(errno));
       return ;
     }
   }

   WRITE_DEBUG("accept_callback() PORT:%d",sin.sin_port);
   dispath_conn(newfd,sin);
}

int send_data(struct ev_loop *loop,CONN_ITEM *c,void *data,unsigned int len)
{
   if  (NULL == c){
     WRITE_ERROR("CONN_ITEM is NULL");
     return -1;
   }

   int ret = 0;
   int nsend = 0;
   int totalsend = 0;
   if (c->wlastpos == 0){/*try sending data immediately*/
     while(totalsend<len){
        nsend = send(c->sfd,(char *)data+totalsend,len-totalsend,0);
        if (nsend == 0 || (nsend == -1 && !(errno == EAGAIN ||errno == EWOULDBLOCK))){
          WRITE_DEBUG("remote socket closed!socket %d [%s:%d]",c->sfd,c->clientip,c->port);
          free_connection(c);
          return -2;
        }else if (nsend>0){
          totalsend+=nsend;
        }else{
          break;
        }
     }
   }else{/*save to buffer tempermently*/

     if (totalsend<len){
         int requiredspace = c->wlastpos + len - totalsend ;
         if (requiredspace > c->wbuffersize){
            c->writebuffer = realloc(c->writebuffer,requiredspace);
            if (NULL == c->writebuffer){
               WRITE_INFO("realloc space failed");
               return -1;
            }
            c->wbuffersize = requiredspace ;
         }
       
         c->wlastpos+=(len - totalsend);
         memcpy(c->writebuffer+c->wlastpos,data+totalsend,len-totalsend);
     }
     
   }

   return totalsend;
}


void recv_callback(struct ev_loop *loop, ev_io *w, int revents)
{
   WORK_THREAD *wt = (WORK_THREAD *)ev_userdata(loop);
   if (NULL == wt){
     WRITE_ERROR("%lu recv_callback() ev_getuserdata() failed,exit",pthread_self());
     return ;
   }

   CONN_ITEM *c = get_conn_byfd(w->fd,wt);
   if (NULL == c){
     WRITE_ERROR("recv_callback() CONN_ITEM is NULL");
     return ;
   }

   int ret =0,last = 0;
   int wants = c->rbuffersize ;
   int remain ;
loop:
   
   pthread_mutex_lock(&c->conn_locker);
   remain = wants - c->rlastpos - 1;
   if (remain<=0) {
      pthread_mutex_unlock(&c->conn_locker);
      return;
   }
   ret=recv(w->fd,c->readbuffer+c->rlastpos,remain,0);
   WRITE_DEBUG("w->fd:%d,c->sfd:%d,c->rlastpos:%d remain:%d ret=%d",w->fd,c->sfd,c->rlastpos,remain,ret);
   if(ret > 0)
   {
      c->rlastpos+=ret;
      ev_io_stop(loop,w);
      pthread_mutex_unlock(&c->conn_locker);
        
      PQ_ITEM *lpNewitem = calloc(1,sizeof(PQ_ITEM)); 
      lpNewitem->sfd = w->fd;
      pthread_mutex_lock(&wt->pending_queue_locker);
      PQ_ITEM *p = TAILQ_FIRST(&wt->pending_queue);
      TAILQ_INSERT_TAIL(&wt->pending_queue,lpNewitem,entries);
      if (NULL == p)
        pthread_cond_signal(&wt->pending_queue_cond);
      pthread_mutex_unlock(&wt->pending_queue_locker);
      WRITE_DEBUG("fd:%d pending queue inserted",c->sfd);
   }
   else if(ret ==0)
   {
     pthread_mutex_unlock(&c->conn_locker);
     WRITE_DEBUG("thread[%lu] remote socket closed!socket fd: %d\n",pthread_self(),c->sfd);
     free_connection(c);
     return;
   }
   else
   {
     if(errno == EAGAIN ||errno == EWOULDBLOCK)
     {
       pthread_mutex_unlock(&c->conn_locker);
       WRITE_DEBUG("c->sfd:%d,errno:%d wait read for next time",c->sfd,errno);
       goto loop;
     }
     else
     {
       pthread_mutex_unlock(&c->conn_locker);
       WRITE_DEBUG("thread[%lu] ret :%d ,close socket fd : %d\n",pthread_self(),ret,w->fd);
       free_connection(c);
       return;
     }
   }
}



void write_callback(struct ev_loop *loop, ev_io *w, int revents)
{
  WRITE_DEBUG("\n\n=====Enter write_calback======\n\n");
  WORK_THREAD *wt = (WORK_THREAD *)ev_userdata(loop);
  CONN_ITEM *c = get_conn_byfd(w->fd,wt);
  if (NULL == c) return ;

  pthread_mutex_lock(&c->conn_locker);
  WRITE_DEBUG("\n\n=====Enter write_calback send_data======\n\n");
  send_data(loop,c,c->writebuffer,c->wlastpos);
  pthread_mutex_unlock(&c->conn_locker);
  
  if (c->wlastpos == 0){
    ev_io_init(&c->io_watcher,recv_callback,c->sfd,EV_READ);
    ev_io_start(loop,&c->io_watcher);
  }
}

 

void signal_callback(struct ev_loop *loop,ev_signal *w,int revents)
{
   int i= 0;
   WRITE_DEBUG("enter signal_callback");
   for (i = 0;i<server->workthread_num;i++){
     WRITE_DEBUG("ev_async_send(%lu)",work_threads[i].thread_id);
     ((WORK_THREAD *)work_threads[i].async_watcher.data)->ifexit = 1;
     ev_async_send(work_threads[i].loop,&(work_threads[i].async_watcher));
   } 

   while (exitednum != server->workthread_num * 2){
     sleep(1);
   }

   WRITE_DEBUG("exitednum:%d == threadnum:%d,exit server",exitednum,server->workthread_num);
   ev_break(loop,EVBREAK_ALL);
}
